package com.jyf.redis.common.utils;

import com.alibaba.fastjson.JSONObject;
//import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import com.jyf.redis.common.config.MyJedisPoolConfig;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.params.XAddParams;
import redis.clients.jedis.params.XReadGroupParams;
import redis.clients.jedis.params.XReadParams;
import redis.clients.jedis.resps.StreamEntry;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.IntStream;

/**
 * 使用jedis实现监听stream消息
 */
@SpringBootTest
@RunWith(SpringRunner.class)
public class JedisStreamMQTest {

    private static final Logger logger = LoggerFactory.getLogger(JedisStreamMQTest.class);

//    public static void main(String[] args) {
    @Test
    public void testRedisMQ(){
        //以下内容根据自己的情况进行修改
        String host = "localhost";
        int port = 6379;
        int timeout = 1000;
        String password = "redis123";
        int database = 0;

        String streamKeyName = "streamtest";
        String groupName = "testgroup";

        String[]consumerNames = {"huhailong", "xiaohu"};

        String listenerType = "DEFAULT";  //GROUP or DEFAULT

        //创建 redis 连接池实例(有报错，废弃)
//        JedisPool pool = new JedisPool(new GenericObjectPoolConfig<>(),host,port,timeout,password,database);
        //静态内部类保存一个jedisPool的实例
        JedisPool pool = MyJedisPoolConfig.getJedisPool();

        JedisStreamMQTest test = new JedisStreamMQTest();
        test.createGroup(pool,streamKeyName,groupName); //创建群组

        if("GROUP".equals(listenerType)){
            test.listenerByGroup(pool,streamKeyName,groupName,consumerNames);   //使用群组和消费者监听
        }else{
            test.listenerDefault(pool,streamKeyName);
        }

        new Thread(()->{    //线程3：用于写入stream数据
            Jedis jedis = pool.getResource();
            while(true) {
                try {
                    Thread.sleep(1500L);
                    Map<String,String> map = new HashMap<>();
                    map.put("currentTime", LocalDateTime.now().toString());
                    jedis.xadd(streamKeyName,map, XAddParams.xAddParams());
                } catch (Exception e) {
                    logger.error(e.getMessage());
                }
            }
        }).start();
    }

    /**
     * 使用群组和消费者监听，该监听可以确保消息不会重复消费，因为每个组每个用户只会消费一次消息
     * @param keyName   stream 名称
     * @param groupName 群组名称
     * @param consumerNames 消费者名称集合
     */
    private void listenerByGroup(JedisPool pool, String keyName, String groupName, String...consumerNames){
        Map<String, StreamEntryID> entryIDMap = new HashMap<>();
        entryIDMap.put(keyName,StreamEntryID.UNRECEIVED_ENTRY);
        //以下为了演示简单就不用线程池了，直接创建两个线程说明问题
        IntStream.range(0,2).forEach(i->{
            Jedis jedis = pool.getResource();   //创建jedis实例
            new Thread(()->{
                while(true){
                    try{
                        Thread.sleep(500L);
                        //下面的 xreadGroup 方法等同与redis中的xreadgroup命令，将block阻塞时间设置为0表示一直阻塞知道收到消息，然后上面StreamEntryID设置为接收最新值
                        List<Map.Entry<String, List<StreamEntry>>> entries = jedis.xreadGroup(groupName, consumerNames[i], XReadGroupParams.xReadGroupParams().block(0), entryIDMap);
                        logger.info("Thread:-{},result:{}",Thread.currentThread().getName(), JSONObject.toJSONString(entries.get(0).getValue().get(0).getFields()));
//                        jedis.xack(keyName,groupName,entries.get(0).getValue().get(0).getID()); //确认消息
                        jedis.xdel(keyName,entries.get(0).getValue().get(0).getID());   //删除消息
                    } catch (Exception e){
                        logger.error(e.getMessage());
                    }
                }
            }).start();
        });

    }

    /**
     * 不使用群组和消费者的概念读取，多个线程会重复消费数据
     * @param keyName stream 名称
     */
    private void listenerDefault(JedisPool pool, String keyName){
        Map<String, StreamEntryID> entryIDMap = new HashMap<>();
        entryIDMap.put(keyName,StreamEntryID.LAST_ENTRY);
        //以下为了演示简单就不用线程池了，直接创建两个线程说明问题
        IntStream.range(0,2).forEach(i->{
            new Thread(()->{
                Jedis jedis = pool.getResource();   //创建jedis实例
                while(true){
                    try{
                        Thread.sleep(500L);
                        List<Map.Entry<String, List<StreamEntry>>> entries = jedis.xread(XReadParams.xReadParams().block(0), entryIDMap);
                        logger.info("Thread:-{},result:{}",Thread.currentThread().getName(), JSONObject.toJSONString(entries.get(0).getValue().get(0).getFields()));
                        jedis.xdel(keyName,entries.get(0).getValue().get(0).getID());
                    } catch (Exception e){
                        logger.error(e.getMessage());
                    }
                }
            }).start();
        });
    }

    private void createGroup(JedisPool pool, String keyName, String groupName){
        Jedis jedis = pool.getResource();
        try{
            //StreamEntryID 表示创建群组并接收新的消息，这里可以根据自己需要设置，0表示读取所有历史消息，后面的boolean值表示如果stream不存在是否创建stream
            jedis.xgroupCreate(keyName,groupName,StreamEntryID.LAST_ENTRY,true);
        } catch (Exception e){
            //这里捕获异常的原因是可能创建时群组已经存在
            logger.error(e.getMessage());
        }
    }
    private static JedisPoolConfig getConfig() {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        //========= jedisPool的一些配置=============================
        poolConfig.setMaxTotal(10000);//最大连接数
        poolConfig.setMaxIdle(50);//最多空闲数
        poolConfig.setMaxWaitMillis(5 * 1000);//当池中没有连接时，最多等待5秒
        return poolConfig;
    }
}
